package com.jhws.core.mns.service;

import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.model.Message;
import com.aliyun.mns.model.QueueMeta;
import com.jhws.core.mns.bo.MNSMessage;
import com.jhws.core.mns.utils.MNSClientUtils;
import com.jhws.util.LogHelper;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;


@Service
public class AliMNSService implements IMNSService {

    @Override
    public void createQueue(String queueName, int pollingWaitSeconds, Long retentionPeriod) throws Exception {
        QueueMeta meta = new QueueMeta(); //生成本地QueueMeta属性，有关队列属性详细介绍见https://help.aliyun.com/document_detail/27476.html
        meta.setQueueName(queueName);  // 设置队列名
        meta.setPollingWaitSeconds(pollingWaitSeconds);
        meta.setMaxMessageSize(2048L);
        if (retentionPeriod >= 60L) {
            meta.setMessageRetentionPeriod(retentionPeriod);
        }
        CloudQueue queue = MNSClientUtils.instance().getClient().createQueue(meta);
        LogHelper.d("--------->   " + queue.getQueueURL());
    }


    @Override
    public void deleteQueue(String queueName) throws Exception {
        //Delete Queue
        CloudQueue queue = MNSClientUtils.instance().getClient().getQueueRef(queueName);
        queue.delete();
    }

    @Override
    public void producer(String queueName, String message) throws Exception {
        CloudQueue queue = MNSClientUtils.instance().getClient().getQueueRef(queueName);
        Message msg = new Message();
        msg.setMessageBody(message);
        Message putMsg = queue.putMessage(msg);
        LogHelper.d("Send message id is: " + putMsg.getMessageBodyAsString());
    }

    @Override
    public MNSMessage comsumer(String queueName) throws Exception {
        CloudQueue queue = MNSClientUtils.instance().getClient().getQueueRef(queueName);// replace with your queue name
        Message popMsg = queue.popMessage();
        MNSMessage mnsMessage = null;
        if (popMsg != null) {
            mnsMessage = new MNSMessage(popMsg.getMessageId(),
                    popMsg.getMessageBodyAsString(),
                    popMsg.getDequeueCount(),
                    popMsg.getEnqueueTime().getTime(),
                    popMsg.getFirstDequeueTime().getTime());
            LogHelper.d("---------->    " + mnsMessage.toString());
            //remember to  delete message when consume message successfully.
            queue.deleteMessage(popMsg.getReceiptHandle());
            LogHelper.d("############# delete message successfully.\n");
        }
        return mnsMessage;
    }

    @Override
    public List<MNSMessage> comsumerList(String queueName, int batchSize) throws Exception {
        CloudQueue queue = MNSClientUtils.instance().getClient().getQueueRef(queueName);// replace with your queue name
        List<Message> messages = queue.batchPopMessage(batchSize);
        List<MNSMessage> mnsMessages = null;
        List<String> receiptHandles;
        MNSMessage mnsMessage = null;
        if (messages != null && messages.size() > 0) {
            LogHelper.d("---------->  messages.size  " + messages.size());
            receiptHandles = new ArrayList<>();
            mnsMessages = new ArrayList<>();
            for (Message message : messages) {
                mnsMessage = new MNSMessage(message.getMessageId(),
                        message.getMessageBodyAsString(),
                        message.getDequeueCount(),
                        message.getEnqueueTime().getTime(),
                        message.getFirstDequeueTime().getTime());
                mnsMessages.add(mnsMessage);
                receiptHandles.add(message.getReceiptHandle());
                LogHelper.d("---------->    " + mnsMessage.toString());
            }
            queue.batchDeleteMessage(receiptHandles);
            //remember to  delete message when consume message successfully.
            LogHelper.d("############# delete message successfully.\n");
        }
        return mnsMessages;
    }
}
